package net.flyinggroup.kafka.handler;

import lombok.extern.slf4j.Slf4j;
import net.flyinggroup.kafka.event.ProductPriceChangedEvent;
import net.flyinggroup.kafka.repository.ProductRepository;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.transaction.Transactional;

@Component
@Transactional
@Slf4j
public class ProductPriceChangedEventHandler {
    private final ProductRepository repository;

    public ProductPriceChangedEventHandler(ProductRepository repository) {
        this.repository = repository;
    }

    @KafkaListener(topics = "product-price-changes", groupId = "demo")
    public void handle(ProductPriceChangedEvent event) {
        log.info("Received a ProductPriceChangedEvent with productCode: {}", event.productCode());
        repository.updateProductPrice(event.productCode(), event.price());
    }
}
